package com.timing.finder.framework;

import com.example.autojob.logging.model.producer.AutoJobLogHelper;
import com.example.autojob.skeleton.model.executor.IMethodObjectFactory;
import com.example.autojob.skeleton.model.task.method.MethodTask;
import com.timing.finder.util.StringUtils;
import com.timing.finder.util.thread.SyncHelper;
import com.timing.finder.util.thread.TimerThreadPoolExecutorHelper;
import lombok.extern.slf4j.Slf4j;
import net.jodah.expiringmap.internal.NamedThreadFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 基于Yaml配置启动的定时加载任务
 *
 * @author JingGe(* ^ ▽ ^ *)
 * @date 2023-03-03 15:34
 * @email 1158055613@qq.com
 */
@Slf4j
public class YamlConfigTimingJob<T extends TimeSeriesData> implements DataLoadTimingJob {
    private final TimingFinderConfig config;
    private final TimeSeriesDataLoader<T> loader;
    private final TimeSeriesDataHandler<T> handler;
    private final TimerThreadPoolExecutorHelper executorHelper;
    private final AtomicBoolean stop = new AtomicBoolean(false);
    /**
     * 本次加载总数
     */
    private final AtomicLong total = new AtomicLong(0L);
    /**
     * 每轮加载总数
     */
    private final AtomicLong preTotal = new AtomicLong(0);
    private final AtomicInteger barrier;
    private final AtomicBoolean withError = new AtomicBoolean(false);
    private final Map<Integer, Boolean> pageSuccess = new ConcurrentHashMap<>();
    private final Map<Integer, AtomicInteger> rollbackCount = new ConcurrentHashMap<>();

    public YamlConfigTimingJob(TimingFinderConfig config, TimeSeriesDataLoader<T> loader, TimeSeriesDataHandler<T> handler) {
        this.config = config;
        this.loader = loader;
        this.handler = handler;
        barrier = new AtomicInteger(0);
        executorHelper = TimerThreadPoolExecutorHelper
                .builder()
                .setInitialCoreTreadCount(config.concurrencyThreadCount)
                .setInitialMaximizeTreadCount(config.concurrencyThreadCount)
                .addTimerEntry(config.cronExpression, config.concurrencyThreadCount, config.concurrencyThreadCount, 60, TimeUnit.SECONDS)
                .addTimerEntry(config.getLowPowerCron(), config.lowPowerThreadCount, config.lowPowerThreadCount, 60, TimeUnit.SECONDS)
                .setThreadFactory(new NamedThreadFactory(config.applicationName + "-loader-%d"))
                .build();

    }

    @Override
    public boolean enable() {
        return config.getEnableLoader();
    }

    @Override
    public String cron() {
        return config.getCronExpression();
    }

    @Override
    public void run(String applicationName, Boolean enable) {
        AutoJobLogHelper logHelper = AutoJobLogHelper.getInstance();
        try {
            logHelper.setSlf4jProxy(log);
            if (enable != null) {
                if (!enable) {
                    logHelper.info("应用{}暂未开启数据加载", applicationName);
                    return;
                }
            } else if (!enable()) {
                logHelper.info("应用{}暂未开启数据加载", applicationName);
                return;
            }
            if (config.clearBeforeRun) {
                long count = handler.clear(String.format("*%s*", config.applicationName));
                logHelper.info("本次清空桶{}个", count);
            }
            config.setStarID(DataLoadTimingJobContext
                    .getInstance()
                    .getCurrentID(config.applicationName));
            logHelper.info("==================================>应用{}数据加载开始，起始ID{}，截止ID{}", applicationName, config.getStarID(), config.getEndID());
            long start = System.currentTimeMillis();
            int threadCount = (config.enableConcurrency ? config.concurrencyThreadCount : 1);
            int pageNum = 1;
            boolean currentBatchSuccess = true;
            do {
                preTotal.set(0);
                barrier.set(0);
                pageSuccess.clear();
                currentBatchSuccess = true;
                for (int i = 1; i <= threadCount; i++) {
                    pageSuccess.put(pageNum, false);
                    Runnable work = loadWork(pageNum++);
                    if (config.enableConcurrency) {
                        executorHelper.submit(work);
                    } else {
                        work.run();
                    }
                }
                SyncHelper.aWait(() -> barrier.get() == threadCount);
                for (Map.Entry<Integer, Boolean> entry : pageSuccess.entrySet()) {
                    if (!entry.getValue()) {
                        if (!rollbackCount.containsKey(entry.getKey())) {
                            rollbackCount.put(entry.getKey(), new AtomicInteger(1));
                        } else {
                            rollbackCount
                                    .get(entry.getKey())
                                    .incrementAndGet();
                        }
                        if (rollbackCount
                                .get(entry.getKey())
                                .get() > 5) {
                            logHelper.warn("第{}页数据回滚超过5次，不再回滚", entry.getKey());
                            continue;
                        }
                        currentBatchSuccess = false;
                        logHelper.warn("第{}页数据保存失败，回滚", entry.getKey());
                        pageNum = Math.min(pageNum, entry.getKey());
                    }
                }
                if (!currentBatchSuccess) {
                    continue;
                }
                DataLoadTimingJobContext
                        .getInstance()
                        .update(applicationName, -1, preTotal.get());
                logHelper.info("应用{}已加载{}条，总计加载{}条", applicationName, total.get(), DataLoadTimingJobContext
                        .getInstance()
                        .getLoadedCount(applicationName));
                handler
                        .cacheRepository()
                        .set(TimingFinderConstant.STATUS_KEY + ":" + config.applicationName, DataLoadTimingJobContext
                                .getInstance()
                                .toJSON(applicationName));
                if (config.maximumExecutionSize > 0 && total.get() > config.maximumExecutionSize) {
                    logHelper.info("应用已加载{}/{}条，总计加载{}条，退出任务", total.get(), config.maximumExecutionSize, DataLoadTimingJobContext
                            .getInstance()
                            .getLoadedCount(applicationName));
                    break;
                }
                BloomFilterContainer
                        .getInstance()
                        .getBloomFilter(applicationName)
                        .storeInCache();
            } while (!stop.get() && !withError.get());
            if (withError.get()) {
                throw new RuntimeException("数据加载时发生异常");
            }
            logHelper.info("==================================>应用{}的数据加载结束，共计加载{}条，用时{}ms", applicationName, DataLoadTimingJobContext
                    .getInstance()
                    .getLoadedCount(applicationName), System.currentTimeMillis() - start);
        } catch (InterruptedException e) {
            logHelper.warn("任务被强制停止");
        } finally {
            stop.set(false);
            withError.set(false);
            total.set(0);
        }
    }

    @SuppressWarnings("unchecked")
    private Runnable loadWork(int pageNum) {
        return () -> {
            AutoJobLogHelper logHelper = AutoJobLogHelper.getInstance();
            logHelper.setSlf4jProxy(log);
            try {
                logHelper.info("线程{}开始load第{}批数据", Thread
                        .currentThread()
                        .getName(), pageNum);
                List<T> data = loader.load(pageNum);
                if (data == null || data.size() == 0) {
                    logHelper.info("线程{}检测到已load完成", Thread
                            .currentThread()
                            .getName());
                    stop.set(true);
                    pageSuccess.put(pageNum, true);
                    return;
                }
                logHelper.info("线程{}load到{}条数据", Thread
                        .currentThread()
                        .getName(), data.size());
                Map<String, List<T>> classified = new HashMap<>();
                data.forEach(item -> {
                    String key = item.key();
                    if (StringUtils.isEmpty(key)) {
                        logHelper.warn("空key");
                        return;
                    }
                    key = key.replace(" ", "");
                    if (!classified.containsKey(key)) {
                        classified.put(key, new ArrayList<>());
                    }
                    classified
                            .get(key)
                            .add(item);
                });
                long start = System.currentTimeMillis();
                int count = 0;
                try {
                    count = handler.storeInCache(classified);
                } catch (DataSaveException e) {
                    boolean testSuccess = false;
                    if (config.testAfterWhenError) {
                        Random random = new Random();
                        TimingFinderApplication<T> application = (TimingFinderApplication<T>) TimingFinderApplicationContext
                                .getInstance()
                                .getCurrentApplicationByName(config.applicationName);
                        TimeSeriesDataFinder<T> finder = application.getTimeSeriesDataFinder();
                        testSuccess = true;
                        for (int i = 0; i < 5; i++) {
                            int index = random.nextInt(data.size());
                            T test = data.get(index);
                            if (finder != null && finder.index(test.key(), test.timestamp(), true, application.handleType.getCacheDataType()) == null) {
                                testSuccess = false;
                            }
                        }
                        if (testSuccess) {
                            logHelper.info("线程{}测试成功", Thread
                                    .currentThread()
                                    .getName());
                        } else {
                            logHelper.warn("线程{}测试失败", Thread
                                    .currentThread()
                                    .getName());
                        }
                    }
                    pageSuccess.put(pageNum, false);
                    logHelper.warn("线程{}监视到第{}页数据保存异常", Thread
                            .currentThread()
                            .getName(), pageNum);
                    return;

                }
                pageSuccess.put(pageNum, true);
                DataLoadTimingJobContext
                        .getInstance()
                        .update(config.applicationName, Objects
                                .requireNonNull(data
                                        .stream()
                                        .max((o1, o2) -> (int) (o1.id() - o2.id()))
                                        .orElse(null))
                                .id());
                logHelper.info("线程{}load完{}条数据用时{}ms", Thread
                        .currentThread()
                        .getName(), count, System.currentTimeMillis() - start);
                total.addAndGet(count);
                preTotal.addAndGet(count);
            } catch (Exception e) {
                if (e instanceof InterruptedException || e
                        .getMessage()
                        .contains("interrupted")) {
                    logHelper.warn("线程{}被强制停止", Thread
                            .currentThread()
                            .getName());
                    stop.set(true);
                    withError.set(false);
                } else {
                    e.printStackTrace();
                    logHelper.error("线程{}执行时发生异常：{}", Thread
                            .currentThread()
                            .getName(), e.getMessage());
                    withError.set(true);
                }
            } finally {
                barrier.incrementAndGet();
            }
        };
    }

    public static class TimingJobObjectFactory implements IMethodObjectFactory {
        @Override
        public Object createMethodObject(MethodTask task, Class<?> methodClass) {
            return TimingFinderApplicationContext
                    .getInstance()
                    .getCurrentApplicationByName((String) task.getParams()[0]).loadTimingJob;
        }
    }
}
